package com.ibm.cps.kafka.listener;

import java.util.HashMap;
import java.util.Map;

import org.apache.log4j.Logger;
import org.apache.storm.guava.base.Throwables;

import com.ibm.cps.forwarder.MessageListener;
import com.ibm.cps.kafka.KafkaTopicConsumeThread;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.newmessage.AbstractDataSourceMetadata;
import com.ibm.cps.newmessage.DataSourceMetadataFactory;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class ForwarderDeleteDataSourceListener extends KafkaTopicConsumeThread {

	private String tenantid;
	private Logger logger;
	private Map<String, MessageListener> ds2Listener;

	public ForwarderDeleteDataSourceListener(String zookeeper, String groupid, String topicName, String tenantid,
			Map<String, MessageListener> ds2Listener) {
		super(zookeeper, groupid, topicName);
		logger = Logger.getLogger(ForwarderDeleteDataSourceListener.class);
		this.ds2Listener = ds2Listener;
		this.tenantid = tenantid;
	}

	@Override
	public void process(byte[] message) throws CPSException {
		String input = new String(message);
		String type;
		try {
			logger.info("Receive metadata in ForwarderDeleteDataSourceListener: " + input);
			HashMap<String, String> key2Value = JSONUtility.getKeyValue(input);
			type = key2Value.get(MessageFields.TYPE);
			logger.info("Type is " + type + " tenantid is " + tenantid);
			if (DataSourceMetadataFactory.isDataSourceType(type)) {
				if (key2Value.get(MessageFields.TENANTID).equals(tenantid)) {
					AbstractDataSourceMetadata dataSource = DataSourceMetadataFactory.parseDataSourceMetadata(input,
							type);
					if (ds2Listener.get(dataSource.getName()) == null) {
						logger.warn("There is no data source " + dataSource.getName() + " to be deleted.");
						return;
					}
					ds2Listener.get(dataSource.getName()).removeDataSource();
					ds2Listener.remove(dataSource.getName());
					logger.info("Data source " + dataSource.getName() + " is deleted.");
				}
			}
		} catch (CPSException e) {
			logger.error(Throwables.getStackTraceAsString(e));
		}
	}
}
